package net.jxtz.jxtzos.landtax;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.PropertyNamingStrategy;
import com.alibaba.fastjson.parser.ParserConfig;
import com.github.shyiko.mysql.binlog.BinaryLogClient;
import com.github.shyiko.mysql.binlog.event.*;
import com.github.shyiko.mysql.binlog.event.deserialization.EventDeserializer;
import lombok.extern.slf4j.Slf4j;
import net.jxtz.jxtzos.entity.landtax.Residential;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.io.Serializable;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;


/**
 * @author a123
 */
@Slf4j
//@Component
public class LandTaxSqlListener implements InitializingBean {

    private static final ParserConfig SNAKE_CASE;

    private static final String SCHEMA = "jxtz-server";
    private static final String TABLE_NAME = "landtax_residential";


    @Value("${spring.datasource.password}")
    private String dbPassword;

    static {
        SNAKE_CASE = new ParserConfig();
        SNAKE_CASE.propertyNamingStrategy = PropertyNamingStrategy.SnakeCase;
    }

    /**
     * @see InitializingBean
     * 继承InitializingBean，在spring启动时执行afterPropertiesSet方法
     */
    @Override
    public void afterPropertiesSet() {
        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(1, 1, 1, TimeUnit.MINUTES, new ArrayBlockingQueue<>(10), (ThreadFactory) Thread::new);
        threadPoolExecutor.execute( () -> {
            // 链接本地数据库
            BinaryLogClient client = new BinaryLogClient("localhost", 3306, SCHEMA,"root", dbPassword);
            EventDeserializer eventDeserializer = new EventDeserializer();
            client.setEventDeserializer(eventDeserializer);
            AtomicBoolean flag = new AtomicBoolean(false);
            boolean connected = client.isKeepAlive();
            log.info("stater binlog listen serve... ");
            log.info("binlog client isKeepAlive ===> " + connected);
            client.registerEventListener(event -> {
                log.info("start: " + JSON.toJSONString(event));
                EventHeader header = event.getHeader();
                EventType eventType = header.getEventType();
                String database = null;
                String table = null;
                // 往下拿到database, table 名称
                if (header.getEventType().equals(EventType.TABLE_MAP)){
                    log.info("table_map ==> " + event.getHeader());
                    String s = JSON.toJSONString(event.getData());
                    database = JSON.parseObject(s).getString("database");
                    table = JSON.parseObject(s).getString("table");
                    log.info("database ==> " + database);
                    log.info("table ==> " + table);
                    if (database.equals(SCHEMA) && table.equals(TABLE_NAME)){
                        // 如果数据库名和表名都等于常量，设置为true
                        flag.set(true);
                    }
                }

                // 往下走，如果flag为true，且为写，修改，删除语句，执行完后flag设置为true，因为要执行下次循环
                if (EventType.isWrite(eventType) && flag.get()) {
                    WriteRowsEventData data = event.getData();
                    log.info("binlog serve to insert ==> " + JSON.toJSONString(data));
                    List<Serializable[]> rows = data.getRows();
                    // 根据binlog文件打印的日志，分析序列化数据，并装载对象返回
                    List<Residential> residentials = BinLogAnalysis.loadObject(rows);
                    syncToMobileServe.addBatch(residentials);
                    flag.set(false);
                } else if (EventType.isUpdate(eventType) && flag.get()) {
                    UpdateRowsEventData data = event.getData();
                    log.info("binlog serve to update ==> " + JSON.toJSONString(data));
                    List<Map.Entry<Serializable[], Serializable[]>> rows = data.getRows();
                    // 修改语句的row会有before和after两条数据，前者是修改之前，后者是修改之后
                    for (Map.Entry<Serializable[], Serializable[]> row : rows) {
                        log.info("before ==> " + Arrays.toString(row.getKey()));
                        log.info("after ==> " + Arrays.toString(row.getValue()));
                        Residential residential = BinLogAnalysis.loadObject(row.getValue());
                        // 一般修改都是单个修改，基本没有批量修改
                        syncToMobileServe.update(residential);
                    }
                    flag.set(false);
                }
            });
            try {
                client.connect();
            } catch (IOException e) {
                e.printStackTrace();
            }
        });

    }


    private final SyncToMobileServe syncToMobileServe;

    public LandTaxSqlListener(SyncToMobileServe syncToMobileServe) {
        this.syncToMobileServe = syncToMobileServe;
    }


}
